package master

import (
	"distributed_db/utils"
	"distributed_db/utils/db"
	"fmt"
	"log"
	"strconv"
	"strings"
	"sync"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

// fetchRegions 通过数据库连接动态获取 Region Server 对应的 Region 列表
func (master *Master) fetchRegions() error {
	for i, rs := range master.regionServers {
		// 通过 GetRegions 函数获取对应 Region Server 中的所有 Region
		err := func() error {
			args := utils.GetAllRegionsArgs{}
			var reply utils.GetAllRegionsReply
			rs.Conn.Call("RegionServerConnection.GetAllRegions", &args, &reply)
			regions := reply.Regions
			if reply.Err != nil {
				return reply.Err
			}
			// 构造 region Server 结构
			master.regionServers[i].Regions = regions
			// 用于去重的 HashMap
			visited := make(map[string]bool)
			// 构造 Table 结构
			for _, region := range regions {
				// 解析 Region 名称，提取 RegionName 和 Region ID
				parts := strings.Split(region.RegionName, "_region_")
				if len(parts) != 2 {
					log.Printf("Invalid region name: %s", region.RegionName)
					continue
				}
				tableName := parts[0]

				// 更新 tables
				master.tablesMutex.Lock()
				table, exists := master.tables[tableName]
				if !exists {
					table = &Table{
						TableName: tableName,
						RecordNum: 0,
						Regions:   []RegionMapping{},
					}
					master.tables[tableName] = table
				}
				table.Regions = append(table.Regions, RegionMapping{
					RegionName:     region.RegionName,
					RecordNum:      region.RecordNum,
					RegionServerID: rs.ID,
				})
				// 一个 RegionName 对应三个副本，因此需要进行去重
				if !visited[region.RegionName] {
					visited[region.RegionName] = true
					table.RecordNum += region.RecordNum
				}
				master.tablesMutex.Unlock()
			}
			return nil
		}()
		if err != nil {
			return err
		}
	}
	return nil
}

// AddRegionServer 添加新的 RegionServer 并同步到文件
func (master *Master) AddRegionServer(ip, port string) error {
	master.regionServersMutex.Lock()
	defer master.regionServersMutex.Unlock()

	// 检查 RegionServer 是否已存在
	for _, server := range master.regionServers {
		if server.IP == ip && server.Port == port {
			fmt.Printf("RegionServer %s:%s already exists\n", ip, port)
			return nil
		}
	}

	// 尝试连接 Region Server，确保连接可用
	conn := utils.MakeEnd(fmt.Sprintf("%s:%s", ip, port))
	if conn == nil {
		return fmt.Errorf("failed to connect to %s:%s", ip, port)
	}

	// 删除新添加 RegionServer 中的所有 Region
	args := utils.DropAllRegionsArgs{}
	var reply utils.DropAllRegionsReply
	conn.Call("RegionServerConnection.DropAllRegions", &args, &reply)
	if reply.Err != nil {
		return reply.Err
	}

	// 添加新的 RegionServer 到内存
	newServer := RegionServer{
		ID:   generateUniqueID(), // 生成唯一 ID
		IP:   ip,
		Port: port,
	}
	master.regionServers = append(master.regionServers, newServer)

	// 将新节点添加到一致性哈希环中
	serverKey := fmt.Sprintf("%s:%s", ip, port)
	master.consistentHash.AddNode(serverKey)

	// 异步更新
	master.taskQueue.Enqueue(func() {
		regionServersToFile(master)
	})

	return nil
}

// RemoveRegionServer 删除指定的 RegionServer 节点
func (master *Master) RemoveRegionServer(ip, port string) error {
	master.regionServersMutex.Lock()
	defer master.regionServersMutex.Unlock()

	// 找到需要删除的 RegionServer
	var removedServer RegionServer
	var remainingServers []RegionServer
	for _, server := range master.regionServers {
		if server.IP == ip && server.Port == port {
			removedServer = server
		} else {
			remainingServers = append(remainingServers, server)
		}
	}

	if removedServer.IP == "" && removedServer.Port == "" {
		return fmt.Errorf("RegionServer %s:%s not found", ip, port)
	}

	// 从一致性哈希环中移除节点
	serverKey := fmt.Sprintf("%s:%s", ip, port)
	master.consistentHash.RemoveNode(serverKey)

	// 重新平衡 Region 数据
	err := master.balanceRegions(removedServer, remainingServers)
	if err != nil {
		return fmt.Errorf("failed to balance regions: %v", err)
	}

	// 更新内存中的 RegionServer 列表
	master.regionServers = remainingServers

	// 更新内存中的 Region Servers 列表和 Tables 列表
	err = master.fetchRegions()
	if err != nil {
		fmt.Printf("failed to fetch regions: %v\n", err)
		return err
	}

	// 异步更新
	master.taskQueue.Enqueue(func() {
		regionServersToFile(master)
	})

	return nil
}

// updateRegionServer 更新全局变量 regionServers 中指定 RegionServer 的信息
func (master *Master) updateRegionServer(serverKey string, updatedServer RegionServer) {
	for i, rs := range master.regionServers {
		if fmt.Sprintf("%s:%s", rs.IP, rs.Port) == serverKey {
			master.regionServers[i] = updatedServer
			break
		}
	}
}

// balanceRegions 在删除 RegionServer 后重新平衡 Region 数据
func (master *Master) balanceRegions(removedServer RegionServer, remainingServers []RegionServer) error {
	// 缓存目标服务器的选择结果
	targetServerCache := make(map[string]RegionServer)
	var cacheMutex sync.Mutex // 添加互斥锁保护缓存访问
	// 控制并发线程数量
	var wg sync.WaitGroup
	// 接收返回值集合
	resChan := make(chan error, len(removedServer.Regions))

	for _, region := range removedServer.Regions {
		wg.Add(1)
		go func(removeRegion db.Region) {
			defer wg.Done()
			// 根据该 Region 的 RegionName 找到对应的 Table Name
			parts := strings.Split(removeRegion.RegionName, "_region_")
			if len(parts) != 2 {
				resChan <- fmt.Errorf("invalid region name: %s", removeRegion.RegionName)
				return
			}
			tableName := parts[0]
			// 检查该 Table Name 是否存在
			table, exists := master.tables[tableName]
			if !exists {
				resChan <- fmt.Errorf("table %s does not exist", tableName)
				return
			}
			// 找到该 Region 的其他副本
			var replicas []RegionServer
			for _, server := range remainingServers {
				for _, r := range server.Regions {
					if r.RegionName == removeRegion.RegionName {
						replicas = append(replicas, server)
						break
					}
				}
			}

			if len(replicas) < 2 {
				resChan <- fmt.Errorf("not enough replicas for region %s after removing server", removeRegion.RegionName)
				return
			}

			if len(remainingServers) < 3 {
				resChan <- fmt.Errorf("not enough remainingServers : %d", len(remainingServers))
				return
			}

			// 使用一致性哈希选择新的目标服务器
			// 尝试次数计数器
			var attempts uint32 = 0

			for len(replicas) < 3 && attempts < MaxAttempts {
				serverKey := master.consistentHash.GetNode(removeRegion.RegionName + strconv.Itoa(int(attempts)))
				var targetServer RegionServer
				// 加锁访问缓存
				cacheMutex.Lock()
				targetServer, exists = targetServerCache[serverKey]
				cacheMutex.Unlock()

				if !exists {
					targetServer = findServerByServerKey(remainingServers, serverKey)
					// 加锁更新缓存
					cacheMutex.Lock()
					targetServerCache[serverKey] = targetServer
					cacheMutex.Unlock()
				}

				// 如果目标服务器无效，则继续尝试
				if targetServer.IP == "" && targetServer.Port == "" {
					attempts++
					continue
				}

				// 检查目标服务器是否已包含该 Region
				exists = false
				for _, replica := range replicas {
					if replica.IP == targetServer.IP && replica.Port == targetServer.Port {
						exists = true
						break
					}
				}
				if exists {
					attempts++
					continue // 如果目标服务器已包含该 Region，则跳过
				}
				// 增加并发任务
				wg.Add(1)
				// 并发复制 Region 数据到目标服务器
				go func(target RegionServer, regionName string) {
					defer wg.Done()
					// 从原始 RegionServer 获取数据
					getArgs := utils.GetRegionDataArgs{
						Table: regionName,
					}
					var getReply utils.GetRegionDataReply
					target.Conn.Call("RegionServerConnection.GetRegionData", &getArgs, &getReply)
					if getReply.Err != nil {
						log.Printf("Failed to fetch data from region %s in server %s:%s: %v", regionName, removedServer.IP, removedServer.Port, getReply.Err)
						return
					}
					columns := getReply.Cols
					data := getReply.Data

					// 在目标服务器上创建 Region 并插入数据
					insertArgs := utils.InsertRegionArgs{
						Table:       regionName,
						Columns:     columns,
						PrimaryKeys: table.PrimaryKey,
						Indexes:     table.Indexes,
						Values:      data,
						TxID:        "",
					}
					var insertReply utils.InsertRegionReply
					target.Conn.Call("RegionServerConnection.InsertRegion", &insertArgs, &insertReply)

					if insertReply.Err != nil {
						log.Printf("Failed to insert data into region %s in server %s:%s: %v", regionName, target.IP, target.Port, insertReply.Err)
						return
					}

					// 更新内存中的 Region 列表
					master.regionServersMutex.Lock()
					target.Regions = append(target.Regions, db.Region{RegionName: regionName})
					master.regionServersMutex.Unlock()
				}(targetServer, removeRegion.RegionName)

				replicas = append(replicas, targetServer)
				attempts++
			}

			// 如果尝试了最大次数仍未找到足够的服务器，返回错误
			if len(replicas) < 3 {
				resChan <- fmt.Errorf("failed to find 3 unique servers for region %s after %d attempts", region.RegionName, MaxAttempts)
				return
			}
		}(region)
	}
	// 等待所有的线程完成
	wg.Wait()
	// 关闭管道
	close(resChan)
	// 处理返回值信息，返回对应的报错
	for res := range resChan {
		if res != nil {
			return res
		}
	}
	return nil
}

// insertDataToRegion 插入数据到指定 Region，并返回插入的数据数量
func (master *Master) insertDataToRegion(transactionID string, regionName string,
	columns []string, primaryKeys []string, indexes []string, data [][]interface{}) (int, error) {
	var inserted int
	var err error
	isFound := false
	// 遍历所有副本并插入数据
	for _, server := range master.regionServers {
		for _, region := range server.Regions {
			if region.RegionName == regionName {
				isFound = true
				inserted, err = func() (int, error) {
					// 插入过滤后的数据
					args := utils.InsertRegionArgs{
						TxID:        transactionID,
						Table:       regionName,
						Columns:     columns,
						PrimaryKeys: primaryKeys,
						Indexes:     indexes,
						Values:      data,
					}
					var reply utils.InsertRegionReply
					server.Conn.Call("RegionServerConnection.InsertRegion", &args, &reply)
					return reply.Ans, reply.Err
				}()
				if err != nil {
					return 0, err
				}
			}
		}
	}
	if isFound {
		return inserted, nil
	}
	return 0, fmt.Errorf("region %s not found", regionName)
}

// deleteDataFromRegion 从指定 Region 删除数据
func (master *Master) deleteDataFromRegion(transactionID string, regionName string, condition string) (int, error) {
	// 遍历所有副本并删除数据
	deleted := 0
	for _, server := range master.regionServers {
		for _, region := range server.Regions {
			if region.RegionName == regionName {
				args := utils.DeleteRegionArgs{
					TxID:  transactionID,
					Table: regionName,
					Cond:  condition,
				}
				var reply utils.DeleteRegionReply
				server.Conn.Call("RegionServerConnection.DeleteRegion", &args, &reply)
				if reply.Err != nil {
					return reply.Ans, reply.Err
				}
				deleted += reply.Ans
			}
		}
	}
	return deleted, nil
}

// updateDataInRegion 更新指定 Region 中的数据
func (master *Master) updateDataInRegion(transactionID string, regionName string, clause string, condition string) error {
	// 遍历所有副本并更新数据
	for _, server := range master.regionServers {
		for _, region := range server.Regions {
			if region.RegionName == regionName {
				args := utils.UpdateRegionArgs{
					TxID:   transactionID,
					Table:  regionName,
					Clause: clause,
					Cond:   condition,
				}
				var reply utils.UpdateRegionReply
				server.Conn.Call("RegionServerConnection.UpdateRegion", &args, &reply)
				if reply.Err != nil {
					return reply.Err
				}
			}
		}
	}
	return nil
}

// findDataInRegion 查找指定 Region 中的数据
func (master *Master) findDataInRegion(transactionID string, regionName string, queryRow string, condition string) ([]map[string]interface{}, error) {
	var results []map[string]interface{}
	for _, server := range master.regionServers {
		for _, region := range server.Regions {
			if region.RegionName == regionName {
				args := utils.QueryRegionArgs{
					TxID:  transactionID,
					Table: regionName,
					Row:   queryRow,
					Cond:  condition,
				}
				var reply utils.QueryRegionReply
				server.Conn.Call("RegionServerConnection.QueryRegion", &args, &reply)
				if reply.Err != nil {
					return nil, reply.Err
				}
				results = append(results, reply.Data...)
			}
		}
	}
	return results, nil
}

// addRegion 添加 Region，并使用一致性哈希选择三个合适的 RegionServer 来存放该 Region 的副本
func (master *Master) addRegion(transactionID string, regionName string,
	columns []string, primaryKeys []string, indexes []string, data [][]interface{}) ([]RegionMapping, error) {
	master.regionServersMutex.Lock()
	defer master.regionServersMutex.Unlock()

	if len(master.regionServers) < 3 {
		return nil, fmt.Errorf("not enough RegionServers to store replicas, need at least 3")
	}

	// 使用一致性哈希选择三个不同的 RegionServer
	selectedServers := make(map[string]RegionServer)
	var attempts uint32 = 0 // 记录尝试次数，避免死循环

	for len(selectedServers) < 3 && attempts < MaxAttempts {
		serverKey := master.consistentHash.GetNode(regionName + strconv.Itoa(int(attempts)))
		for _, server := range master.regionServers {
			if fmt.Sprintf("%s:%s", server.IP, server.Port) == serverKey {
				// 确保选择的服务器是唯一的
				if _, exists := selectedServers[serverKey]; !exists {
					selectedServers[serverKey] = server
				}
				break
			}
		}
		attempts++
	}

	// 如果尝试了最大次数仍未找到足够的服务器，返回错误
	if len(selectedServers) < 3 {
		return nil, fmt.Errorf("failed to select 3 unique RegionServers using consistent hash after %d attempts", MaxAttempts)
	}

	// 创建 Region 并插入到选中的 RegionServer 中
	for serverKey, server := range selectedServers {
		err := func() error {
			args := utils.InsertRegionArgs{
				TxID:        transactionID,
				Table:       regionName,
				Columns:     columns,
				PrimaryKeys: primaryKeys,
				Indexes:     indexes,
				Values:      data,
			}
			var reply utils.InsertRegionReply
			// 插入数据到新创建的 Region 中
			server.Conn.Call("RegionServerConnection.InsertRegion", &args, &reply)
			if reply.Err != nil {
				return reply.Err
			}
			return nil
		}()
		if err != nil {
			return nil, fmt.Errorf("failed to add region %s to server %s:%s: %v", regionName, server.IP, server.Port, err)
		}

		// 更新内存中的 Region 列表
		server.Regions = append(server.Regions, db.Region{RegionName: regionName})
		// 使用统一的函数更新全局变量 regionServers
		master.updateRegionServer(serverKey, server)
	}

	dataSize := len(data)
	var regionMaps []RegionMapping
	for _, server := range selectedServers {
		// 使用 RegionServer 的 ID 替代具体信息
		regionMap := RegionMapping{
			RegionName:     regionName,
			RecordNum:      dataSize,
			RegionServerID: server.ID,
		}
		regionMaps = append(regionMaps, regionMap)
	}

	return regionMaps, nil
}

// removeRegion 删除 Region，并确保操作的原子性
func (master *Master) removeRegion(regionName string) error {
	master.regionServersMutex.Lock()
	defer master.regionServersMutex.Unlock()

	// 找到所有包含该 Region 的 RegionServer
	var regionServersWithRegion []RegionServer
	for _, server := range master.regionServers {
		for _, region := range server.Regions {
			if region.RegionName == regionName {
				regionServersWithRegion = append(regionServersWithRegion, server)
				break
			}
		}
	}

	if len(regionServersWithRegion) != 3 {
		return fmt.Errorf("cannot remove region %s, expected 3 replicas but found %d", regionName, len(regionServersWithRegion))
	}

	// 从所有包含该 Region 的 RegionServer 中删除它
	for _, server := range regionServersWithRegion {
		// 在对应数据库中删除对应的 Region
		err := func() error {
			args := utils.DropRegionArgs{
				Region: regionName,
			}
			var reply utils.DropRegionReply
			server.Conn.Call("RegionServerConnection.DropRegion", &args, &reply)
			if reply.Err != nil {
				return reply.Err
			}
			return nil
		}()
		if err != nil {
			return fmt.Errorf("failed to remove region %s from server %s:%s: %v", regionName, server.IP, server.Port, err)
		}

		// 更新内存中的 Region 列表
		for i, region := range server.Regions {
			if region.RegionName == regionName {
				server.Regions = append(server.Regions[:i], server.Regions[i+1:]...)
				break
			}
		}

		// 使用统一的函数更新全局变量 regionServers
		serverKey := fmt.Sprintf("%s:%s", server.IP, server.Port)
		master.updateRegionServer(serverKey, server)
	}

	return nil
}

// findServerByServerKey 根据 serverKey 查找对应的 RegionServer
func findServerByServerKey(servers []RegionServer, serverKey string) RegionServer {
	for _, server := range servers {
		key := fmt.Sprintf("%s:%s", server.IP, server.Port)
		if key == serverKey {
			return server
		}
	}
	return RegionServer{}
}

// generateUniqueID 生成唯一的 ID
func generateUniqueID() string {
	return fmt.Sprintf("rs-%d", time.Now().UnixNano())
}
